package today.rocky.boot.kafka.client.provider;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author Rocky
 * @version 1.0
 * @date 2020/3/18 13:05
 */
public class KafkaProvider extends Thread{

    KafkaProducer<Integer,String> producer ;
    String topic;

    public static void main(String[] args) {
        //new KafkaProvider("test").start();
        new KafkaProvider("partition_test").start();
    }

    public KafkaProvider(String topic){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.48.131:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"rocky-producer");
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"today.rocky.boot.kafka.client.provider.PartitionStrategy");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        this.producer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
    }

    @Override
    public void run() {
        int num = 0;
        String msg = "kafka test msg";
        while (num<20){
            try {
                RecordMetadata recordMetadata = producer.send(new ProducerRecord<>(topic, msg+num)).get();

                System.out.println(recordMetadata.offset()+"->"+recordMetadata.partition()+"->"+recordMetadata.topic());
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
            num++;
        }

    }
}
